package com.kitten.relation_user.biz.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.kitten.framework.common.exception.BizException;
import com.kitten.framework.common.response.PageResponse;
import com.kitten.framework.common.response.Response;
import com.kitten.framework.common.utils.DateUtils;
import com.kitten.framework.common.utils.JsonUtils;
import com.kitten.framework.context.holder.LoginUserContextHolder;
import com.kitten.relation_user.biz.constants.MQConstant;
import com.kitten.relation_user.biz.constants.RedisKeyConstant;
import com.kitten.relation_user.biz.domain.dataobject.FansDO;
import com.kitten.relation_user.biz.domain.dataobject.FollowingDO;
import com.kitten.relation_user.biz.domain.mapper.FansDOMapper;
import com.kitten.relation_user.biz.domain.mapper.FollowingDOMapper;
import com.kitten.relation_user.biz.enums.LuaResultEnum;
import com.kitten.relation_user.biz.enums.ResponseCodeEnum;
import com.kitten.relation_user.biz.model.dto.FollowUserMqDTO;
import com.kitten.relation_user.biz.model.dto.UnfollowUserMqDTO;
import com.kitten.relation_user.biz.model.vo.*;
import com.kitten.relation_user.biz.rpc.UserRpcService;
import com.kitten.relation_user.biz.service.RelationService;
import com.kitten.user.dto.resp.FindUserByIdRspDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Set;

@Slf4j
@Service
public class RelationServiceImpl implements RelationService {
    @Resource
    private UserRpcService userRpcService;
    @Resource
    private FollowingDOMapper followingDOMapper;
    @Resource
    private FansDOMapper fansDOMapper;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;


    /**
     * 关注用户
     * @param followUserReqVO
     * @return
     */
    @Override
    public Response<?> follow(FollowUserReqVO followUserReqVO) {
        Long followUserId = followUserReqVO.getFollowUserId();  // 被关注用户 ID
        // 当前登录用户ID
        Long userId = LoginUserContextHolder.getUserId();
        if (Objects.equals(userId, followUserId)) {
            throw new BizException(ResponseCodeEnum.CANT_FOLLOW_YOUR_SELF);
        }
        // : 校验关注的用户是否存在 调用用户rpc服务
        FindUserByIdRspDTO findUserByIdRspDTO = userRpcService.findById(followUserId);
        if (Objects.isNull(findUserByIdRspDTO)) {
            throw new BizException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }
        // : Lua 脚本：校验并添加关注关系
        String followingKey = RedisKeyConstant.buildUserFollowingKey(userId);       // 当前用户关注列表的 Redis key  如: folllowing:100 表示用户100的关注列表
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_check_and_add.lua")));    // 加载 LUA 脚本
        script.setResultType(Long.class);   // 返回类型是 Long
        LocalDateTime now = LocalDateTime.now();
        long timestamp = DateUtils.localDateTime2Timestamp(now);
        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(followingKey), followUserId, timestamp);

        checkLuaScriptResult(result);

        // ZSET 不存在
        if (Objects.equals(result, LuaResultEnum.ZSET_NOT_EXIST.getCode())) {   // redis中不存在该用户的关注列表
            // 从数据库查询当前用户的关注关系记录
            List<FollowingDO> followingDOS = followingDOMapper.selectByUserId(userId);
            log.info("用户 {} 在数据库中的关注关系记录: {}", userId, followingDOS);

            // 随机过期时间
            // 保底1天+随机秒数
            long expireSeconds = 60*60*12 + RandomUtil.randomInt(60*60*12);

            // 若记录为空，直接 ZADD 对象, 并设置过期时间
            if (CollUtil.isEmpty(followingDOS)) {   // 该用户在数据库中也未关注任何人
                log.info("用户 {} 还没有任何关注, 直接 ZADD ");
                DefaultRedisScript<Long> script2 = new DefaultRedisScript<>();
                script2.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_add_and_expire.lua")));    // 加载 LUA 脚本: 实现关注记录 in redis
                script2.setResultType(Long.class);

                // TODO: 可以根据用户类型，设置不同的过期时间，若当前用户为大V, 则可以过期时间设置的长些或者不设置过期时间；如不是，则设置的短些
                // 如何判断呢？可以从计数服务获取用户的粉丝数，目前计数服务还没创建，则暂时采用统一的过期策略
                redisTemplate.execute(script2, Collections.singletonList(followingKey), followUserId, timestamp, expireSeconds);
            } else { // 若记录不为空，则将关注关系数据全量同步到 Redis 中，并设置过期时间；
                // 构建 Lua 参数
                Object[] luaArgs = buildLuaArgs(followingDOS, expireSeconds);

                // 执行 Lua 脚本，批量同步关注关系数据到 Redis 中
                DefaultRedisScript<Long> script3 = new DefaultRedisScript<>();
                script3.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_batch_add_and_expire.lua")));
                script3.setResultType(Long.class);
                redisTemplate.execute(script3, Collections.singletonList(followingKey), luaArgs);

                // 再次调用上面的 Lua 脚本：follow_check_and_add.lua , 将最新的关注关系添加进去
                result = redisTemplate.execute(script, Collections.singletonList(followingKey), followUserId, timestamp);
                checkLuaScriptResult(result);
            }
        }
        // TODO: 发送 MQ
        // 1. 构建消息体DTO
        FollowUserMqDTO followUserMqDTO = FollowUserMqDTO.builder()
                .userId(userId)
                .followUserId(followUserId)
                .createTime(now)
                .build();
        // 2. 消息体转JSON
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(followUserMqDTO)).build();
        String destination = MQConstant.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstant.TAG_FOLLOW;
        String hashkey = String.valueOf(userId);
        // 3. 发送消息
        log.info("====> 发送关注消息到MQ: destination={}, message={}", destination, message);
        rocketMQTemplate.asyncSendOrderly(destination, message, hashkey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("====> 关注消息发送成功: sendResult={}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("====> 关注消息发送异常: throwable={}", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 取关业务
     * @param unfollowUserReqVO
     * @return
     */
    @Override
    public Response<?> unfollow(UnfollowUserReqVO unfollowUserReqVO) {
        Long unfollowUserId = unfollowUserReqVO.getUnfollowUserId();
        // 当前登录用户id
        Long userId = LoginUserContextHolder.getUserId();

        if (Objects.equals(userId, unfollowUserId)) {   // 不能取关自己
            throw new BizException(ResponseCodeEnum.CANT_UNFOLLOW_YOUR_SELF);
        }
        // 校验取关的用户是否存在
        FindUserByIdRspDTO findUserByIdRspDTO = userRpcService.findById(unfollowUserId);
        if (Objects.isNull(findUserByIdRspDTO)) {
            throw new BizException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }

        // redis: 校验关注列表是否存在, 列表中是否存在 unfollowUserId, 若存在则删除
        String followingKey = RedisKeyConstant.buildUserFollowingKey(userId);   // 当前用户的关注列表Key
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/unfollow_check_and_delete.lua")));
        script.setResultType(Long.class);

        Long result = redisTemplate.execute(script, Collections.singletonList(followingKey), unfollowUserId);
        // 校验 Lua 脚本执行结果
        // 取关的用户不在关注列表中
        if (Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode())) {
            throw new BizException(ResponseCodeEnum.NOT_FOLLOWED_YET);
        }

        if (Objects.equals(result, LuaResultEnum.ZSET_NOT_EXIST.getCode())) { // Redis中不存在 ZSET关注列表
            List<FollowingDO> followingDOS = followingDOMapper.selectByUserId(userId);  // 数据库查询关注记录

            if (CollUtil.isEmpty(followingDOS)) {   // 数据库中不存在关注记录
                throw new BizException(ResponseCodeEnum.NOT_FOLLOWED_YET);
            } else { // 若记录不为空，则将关注关系数据全量同步到 Redis 中，并设置过期时间；
                // 构建 Lua 参数
                long expireSeconds = RandomUtil.randomInt(60 * 60 * 12) + 60 * 60 * 12;
                Object[] luaArgs = buildLuaArgs(followingDOS, expireSeconds);

                // 执行 Lua 脚本，批量同步关注关系数据到 Redis 中
                DefaultRedisScript<Long> script3 = new DefaultRedisScript<>();
                script3.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_batch_add_and_expire.lua")));
                script3.setResultType(Long.class);
                redisTemplate.execute(script3, Collections.singletonList(followingKey), luaArgs);

                // 再次调用上面的 Lua 脚本：unfollow_check_and_delete.lua , 将取关的用户删除
                result = redisTemplate.execute(script, Collections.singletonList(followingKey), unfollowUserId);
                // 再次校验结果
                if (Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode())) {
                    throw new BizException(ResponseCodeEnum.NOT_FOLLOWED_YET);
                }
            }
        }

        // 发送MQ
                // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
                UnfollowUserMqDTO unfollowUserMQDTO = UnfollowUserMqDTO.builder()
                        .userId(userId)
                        .unfollowUserId(unfollowUserId)
                        .createTime(LocalDateTime.now())
                        .build();
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(unfollowUserMQDTO)).build();
        String destination = MQConstant.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstant.TAG_UNFOLLOW;   // 取关消息, 指定Topic

        String hashkey = String.valueOf(userId);
        log.info("====> 发送取关消息到MQ: destination={}, message={}", destination, message);

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, hashkey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("====> 取关消息MQ 发送成功: sendResult={}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("====> 取关消息MQ 发送异常: throwable={}", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 获取用户的关注列表
     * @param findFollowingListReqVO
     * @return
     */
    @Override
    public PageResponse<FindFollowingUserRspVO> findFollowingList(FindFollowingListReqVO findFollowingListReqVO) {
        Long userId = findFollowingListReqVO.getUserId();   // 查询的用户 ID
        Integer pageNo = findFollowingListReqVO.getPageNo();    // 当前页码

        // 1. 从 Redis 中获取关注列表
        String followingListRedisKey = RedisKeyConstant.buildUserFollowingKey(userId);   // 当前用户的关注列表Key
        Long total = redisTemplate.opsForZSet().zCard(followingListRedisKey);   // 获取关注的数目
        // 返参
        List<FindFollowingUserRspVO> findFollowingUserRspVOS = null;
        long limit = 10;
        if (total > 0) {     // 缓存中有数据
            long totalPage = PageResponse.getTotalPage(total, limit); // 每页显示10条, 计算总页数
            if (pageNo > totalPage) {   // 请求的页码 超过总页数
                return PageResponse.success(null, pageNo, total);
            }
            // 从 Redis 查询分页数据
            long offset = (pageNo - 1) * limit;
            // 使用 ZREVRANGEBYSCORE 命令按 score 降序获取元素，同时使用 LIMIT 子句实现分页
            // 注意：这里使用了 Double.POSITIVE_INFINITY 和 Double.NEGATIVE_INFINITY 作为分数范围
            // 因为关注列表最多有 1000 个元素，这样可以确保获取到所有的元素
            Set<Object> followingUserIdSet = redisTemplate.opsForZSet().reverseRangeByScore(followingListRedisKey, Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, offset, limit);
            if (CollUtil.isNotEmpty(followingUserIdSet)) {
                List<Long> userIds = followingUserIdSet.stream().map(object -> Long.valueOf(object.toString())).toList();

                // RPC 批量查询方法
                findFollowingUserRspVOS = rpcUserServiceAndDTO2VO(userIds, findFollowingUserRspVOS);
            }
        } else { // TODO : Redis缓存中没有数据, 需要从数据库中查询
            long count = followingDOMapper.selectCountByUserId(userId);
            long totalPage = PageResponse.getTotalPage(count, 10);  // 计算总页数
            if (pageNo > totalPage) {   // 请求的页码 超过总页数
                return PageResponse.success(null, pageNo, total);
            }
            long offset = PageResponse.getOffset(pageNo, limit);
            // 分页查询
            List<FollowingDO> followingDOS = followingDOMapper.selectPageListByUserId(userId, offset, limit);
            total = count;
            if (CollUtil.isNotEmpty(followingDOS)) {
                List<Long> followingUserIds = followingDOS.stream().map(FollowingDO::getFollowingUserId).toList();
                // RPC 批量查询方法
                findFollowingUserRspVOS = rpcUserServiceAndDTO2VO(followingUserIds, findFollowingUserRspVOS);
                // 缓存关注列表到 Redis
                threadPoolTaskExecutor.submit(() -> {
                    syncFollowingListToRedis(userId);
                });
            }
        }
        return PageResponse.success(findFollowingUserRspVOS, pageNo, total);
    }

    /**
     * 获取用户的粉丝列表
     * @param findFansListReqVO
     * @return
     */
    @Override
    public PageResponse<FindFansUserRspVO> findFansList(FindFansListReqVO findFansListReqVO) {
        Long userId = findFansListReqVO.getUserId();    // 查询的用户 ID
        Integer pageNo = findFansListReqVO.getPageNo();    // 当前页码

        String findFansRedisKey = RedisKeyConstant.buildUserFansKey(userId);

        Long count = redisTemplate.opsForZSet().zCard(findFansRedisKey);    // 获取粉丝数
        // 返参
        List<FindFansUserRspVO> findFansUserRspVOS = null;
        long limit = 10;
        if(count > 0) { // redis中有数据
            long totalPage = PageResponse.getTotalPage(count, limit);
            if (pageNo > totalPage) {   // 请求的页码 超过总页数
                return PageResponse.success(null, pageNo, count);
            }
            long offset = (pageNo - 1) * limit;
            Set<Object> fansUserIdSet = redisTemplate.opsForZSet().reverseRangeByScore(findFansRedisKey, Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, offset, limit);
            if (CollUtil.isNotEmpty(fansUserIdSet)) {
                List<Long> fansUserIds = fansUserIdSet.stream().map(object -> Long.valueOf(object.toString())).toList();
                // RPC 批量查询方法
                findFansUserRspVOS = rpcUserServiceAndCountServiceAndDTO2VO(fansUserIds, findFansUserRspVOS);
            }
        } else { // redis中没有数据, 从数据库中查询
            count = fansDOMapper.selectCountByUserId(userId);
            long totalPage = PageResponse.getTotalPage(count, 10);
            if (pageNo > totalPage) {   // 请求的页码 超过总页数
                return PageResponse.success(null, pageNo, count);
            }
            long offset = PageResponse.getOffset(pageNo, limit);
            List<FansDO> fansDOS = fansDOMapper.selectPageListByUserId(userId, offset, limit);
            if (CollUtil.isNotEmpty(fansDOS)) {
                List<Long> fansUserIds = fansDOS.stream().map(FansDO::getFansUserId).toList();
                // RPC 批量查询方法
                findFansUserRspVOS = rpcUserServiceAndCountServiceAndDTO2VO(fansUserIds, findFansUserRspVOS);
                // 缓存粉丝列表到 Redis
                threadPoolTaskExecutor.submit(() -> {
                    syncFansListToRedis(userId);
                });
            }
        }

        return PageResponse.success(findFansUserRspVOS, pageNo, count);
    }

    /**
     * 校验 Lua 脚本执行结果
     * @param result
     */
    private static void checkLuaScriptResult(Long result) {
        LuaResultEnum luaResultEnum = LuaResultEnum.valueOf(result);

        if (Objects.isNull(luaResultEnum)) throw new RuntimeException("Lua 返回结果错误");
        // 校验 Lua 脚本执行结果
        switch (luaResultEnum) {
            // 关注数已达到上限
            case FOLLOW_LIMIT -> throw new BizException(ResponseCodeEnum.FOLLOWING_COUNT_LIMIT);
            // 已经关注了该用户
            case ALREADY_FOLLOWED -> throw new BizException(ResponseCodeEnum.ALREADY_FOLLOWED);
        }
    }

    /**
     * 构建 Lua 脚本参数
     * @param followingDOS
     * @param expireSeconds
     * @return
     */
    private static Object[] buildLuaArgs(List<FollowingDO> followingDOS, long expireSeconds) {
        int argsLength = followingDOS.size() * 2 + 1; // 每个关注关系有 2 个参数（score 和 value），再加一个过期时间
        Object[] luaArgs = new Object[argsLength];

        int i = 0;
        for (FollowingDO following : followingDOS) {
            luaArgs[i] = DateUtils.localDateTime2Timestamp(following.getCreateTime()); // 关注时间作为 score
            luaArgs[i + 1] = following.getFollowingUserId();          // 关注的用户 ID 作为 ZSet value
            i += 2;
        }

        luaArgs[argsLength - 1] = expireSeconds; // 最后一个参数是 ZSet 的过期时间
        return luaArgs;
    }

    private static Object[] buildFansZsetLuaArgs(List<FansDO> fansDOS, long expireSeconds) {
        int argsLength = fansDOS.size() * 2 + 1; // 每个粉丝关系有 2 个参数（score 和 value），再加一个过期时间
        Object[] luaArgs = new Object[argsLength];
        int i = 0;
        for (FansDO fans : fansDOS) {
            luaArgs[i] = DateUtils.localDateTime2Timestamp(fans.getCreateTime());
            luaArgs[i + 1] = fans.getFansUserId();
            i += 2;
        }
        luaArgs[argsLength - 1] = expireSeconds; // 最后一个参数是 ZSet 的过期时间
        return luaArgs;
    }

    /**
     * 调用 RPC 服务，批量查询用户信息同时转换 DTO 到 VO
     * @param followingUserIds
     * @param findFollowingUserRspVOS
     * @return
     */
    private List<FindFollowingUserRspVO> rpcUserServiceAndDTO2VO(List<Long> followingUserIds, List<FindFollowingUserRspVO> findFollowingUserRspVOS) {
        // RPC: 批量查询用户信息
        List<FindUserByIdRspDTO> findUserByIdRspDTOS = userRpcService.findByIds(followingUserIds);

        // 若不为空，DTO 转 VO
        if (CollUtil.isNotEmpty(findUserByIdRspDTOS)) {
            findFollowingUserRspVOS = findUserByIdRspDTOS.stream()
                    .map(dto -> FindFollowingUserRspVO.builder()
                            .userId(dto.getId())
                            .avatar(dto.getAvatar())
                            .nickname(dto.getNickName())
                            .introduction(dto.getIntroduction())
                            .build())
                    .toList();
        }
        return findFollowingUserRspVOS;
    }

    /**
     * 调用 RPC 服务，批量查询用户信息和计数服务, 同时转换 DTO 到 VO
     * @param fansUserIds
     * @param findFansUserRspVOS
     * @return
     */
    private List<FindFansUserRspVO> rpcUserServiceAndCountServiceAndDTO2VO(List<Long> fansUserIds, List<FindFansUserRspVO> findFansUserRspVOS) {
        List<FindUserByIdRspDTO> findUserByIdRspDTOS = userRpcService.findByIds(fansUserIds);
        // TODO: 调用计数服务，获取粉丝数

        // 若不为空，DTO 转 VO
        if (CollUtil.isNotEmpty(findUserByIdRspDTOS)) {
            findFansUserRspVOS = findUserByIdRspDTOS.stream()
                    .map(dto -> FindFansUserRspVO.builder()
                            .userId(dto.getId())
                            .avatar(dto.getAvatar())
                            .nickname(dto.getNickName())
                            .noteTotal(0L)
                            .fansTotal(0L)
                            .build())
                    .toList();
        }
        return findFansUserRspVOS;
    }

    /**
     * 同步关注列表到 Redis
     */
    private void syncFollowingListToRedis(Long userId){
        List<FollowingDO> followingDOS = followingDOMapper.selectAllByUserId(userId);   // 最多1000
        if (CollUtil.isNotEmpty(followingDOS)) {
            String followingListRedisKey = RedisKeyConstant.buildUserFollowingKey(userId);
            long expireSeconds = RandomUtil.randomInt(60 * 60 * 6) + 60 * 60 * 6;
            Object[] luaArgs = buildLuaArgs(followingDOS, expireSeconds);    // 构建 Lua 参数
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_batch_add_and_expire.lua")));
            script.setResultType(Long.class);
            redisTemplate.execute(script, Collections.singletonList(followingListRedisKey), luaArgs);
        }
    }

    /**
     * 同步粉丝列表到 Redis
     * @param userId
     */
    private void syncFansListToRedis(Long userId){
        List<FansDO> fansDOS = fansDOMapper.select5000FansByUserId(userId);
        if (CollUtil.isNotEmpty(fansDOS)) {
            String userFansListRedisKey = RedisKeyConstant.buildUserFansKey(userId);
            long expireSeconds = RandomUtil.randomInt(60 * 60 * 6) + 60 * 60 * 6;
            Object[] luaArgs = buildFansZsetLuaArgs(fansDOS, expireSeconds);
            // 执行lua
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_batch_add_and_expire.lua")));
            script.setResultType(Long.class);
            redisTemplate.execute(script, Collections.singletonList(userFansListRedisKey), luaArgs);
        }
    }


}
